package com.qf.ability.search.application;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;

/**
 *
 * 准备好一个项目 - 数据分库分表、高并发（秒杀 发红包）、
 *
 * 相关主题的创建
 * description:
 * author: Ken
 * 公众号：Java架构栈
 */
@Configuration
@EnableKafkaStreams
public class KafkaConfig {

    @Autowired
    private StreamsBuilderFactoryBean streamsBuilderFactoryBean;

    @Bean
    public NewTopic getNewTopic(){
        return new NewTopic("hot_word_in", 4, (short) 2);
    }

    @Bean
    public NewTopic getNewTopic2(){
        return new NewTopic("hot_word_out", 4, (short) 2);
    }

//    @Bean
//    public KafkaStreams getKafkaStream(KafkaProperties properties){
//        KafkaStreams kafkaStreams = new KafkaStreams( ,properties.buildStreamsProperties());
//        return kafkaStreams;
//    }

    @Bean
    public KStream getKstream() throws Exception {
        System.out.println("初始化KafkaStream流！");
        StreamsBuilder streamBuilder = streamsBuilderFactoryBean.getObject();
        KTable<Object, Long> kTable = streamBuilder.stream("hot_word_in")
                .map((key, value) -> new KeyValue<>(value, "1".getBytes()))
                .groupByKey()
                //设置KafkaStream流的滑动窗口
//                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .count();

//        kTable.toStream().foreach((key, value) -> {
////            System.out.println("统计的结果：" + new String((byte[]) key.key()) + "--" + value + "--窗口：" + key.window().startTime() + "-" + key.window().endTime());
////        });

        kTable.toStream().map((key, value) -> new KeyValue<>((byte[])key, (value + "").getBytes()))
                .to("hot_word_out", Produced.with(Serdes.ByteArray(), Serdes.ByteArray()));

        return kTable.toStream();
    }
}
